package notify

import (
	"context"
	"enotify/internal/xys/notify/event"
	pkgNotify "enotify/pkg/notify"
	pkgNotifyConf "enotify/pkg/notify/conf"
	"fmt"
	"log"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)
const (
	WX_KEY = "wx"
	SMS_KEY = "sms"
	APP_KEY = "app"
	STATE_INIT = 0
	STATE_PROCESS = 1
	STATE_STOP = 2
	STATE_FINISH = 3
	STATE_FAIL = 4
	MSGTYPE_WX = "1"
	MSGTYPE_APP = "2"
	MSGTYPE_SMS = "3"
)

func init() {
	log.SetFlags(log.Lshortfile | log.LstdFlags)
}

type GroupReq struct {
	Id int64
	SceneId int64
	LoadUserFun LoadingUserInfo
	SyncPrgFun SyncProgress
	Req interface{}
}

type Group struct {
	wg sync.WaitGroup
	stop chan struct{}
	counter map[string]int
	stopReason string
	state int32 // 1 处理中，2 中止， 3 发送完成 4，发送失败
	config *Config
	sender *pkgNotify.Sender
	progress *Progress
	groupId int64
	restart bool
	cache *NotifyCache
}
func NewGroup(conf *Config,cache *NotifyCache) *Group{
	wg := sync.WaitGroup{}
	stop := make(chan struct{},1)
	g := &Group{
		wg: wg,stop: stop,
		counter: map[string]int{},
		config: conf,
		cache: cache,

	}
	senderConfig := &pkgNotifyConf.Config{
		Wx:  conf.Provider.Wx,
		Sms: conf.Provider.Sms,
		App: conf.Provider.App,
	}
	sender := pkgNotify.NewSender(senderConfig)
	if len(senderConfig.Wx)>0{
		wxOp := pkgNotify.SenderWithWxTemplate(senderConfig,g)
	    wxOp(sender)
	}
	if len(senderConfig.App)>0{
		appOp := pkgNotify.SenderWithApp(senderConfig)
		appOp(sender)
	}
	if len(senderConfig.Sms)>0{
		smsOp := pkgNotify.SenderWithSMS(senderConfig,g)
		smsOp(sender)
	}

	g.sender = sender
	return g
}
func (g *Group) Id() (int64,error) {
	return 0,nil
}
func (g *Group) GetAccessToken(appId string) (string,error) {
	return "",nil
}
func (g *Group) startGroup()  {
	atomic.CompareAndSwapInt32(&g.state,STATE_INIT,STATE_PROCESS)
}
func (g *Group) failGroup(reason string)  {
	atomic.StoreInt32(&g.state,STATE_FAIL)
}
func (g *Group) finishGroup()  {
	atomic.CompareAndSwapInt32(&g.state,STATE_PROCESS,STATE_FINISH)
}

func (g *Group) incrSuccessCounter(key string,flags []string)  {
	g.progress.IncrSuccess(key,flags)
}
func (g *Group) incrFailCounter(key string,flags []string)  {
	g.progress.IncrFail(key,flags)

}


func (g *Group) Stop()  {
	g.state = STATE_STOP
	g.stopReason = "系统终止"
	close(g.stop)
	g.wg.Wait()
}

func (g *Group) syncPrg(syncFunc SyncProgress) <-chan struct{}{
	syncFinish := make(chan struct{},0)

	if g.progress.syncQueue == nil {
		close(syncFinish)
		return syncFinish
	}
	batchNum := 10
	go func() {

		syncInfo := SyncInfo{
			State:      0,
			FailReason: "",
			Infos:     nil ,
		}
		sync := func(tmpInfos []NeedSyncPrgInfo) {
			syncInfo.State = int(g.state)
			syncInfo.FailReason = g.stopReason
			syncInfo.Infos = tmpInfos
			syncFunc(syncInfo)
		}

		prgInfos := g.progress.GetProgressInfo()
		initInfo := make([]NeedSyncPrgInfo,0,len(prgInfos))
		cacheMap := map[string]StaticInfo{}
		for _,v := range prgInfos {
			initInfo = append(initInfo,v)
			cacheMap[v.Key] = StaticInfo{
				Success: v.SuccessCt,
				Fail:    v.FailCt,
			}
		}
		sync(initInfo)
		prgInfos = nil
		Infos := make([]NeedSyncPrgInfo,0,batchNum)
		for msg := range g.progress.syncQueue {
			log.Printf("syncQueue msg %+v",msg)
			if _,ok := cacheMap[msg.Key]; !ok {
				cacheMap[msg.Key] = StaticInfo{
					Success: 0,
					Fail:    0,
				}
			}
			Infos = append(Infos, msg)
			if msg.FailCt>cacheMap[msg.Key].Fail {
				f := cacheMap[msg.Key]
				f.Fail = msg.FailCt
				cacheMap[msg.Key] = f

			}
			if msg.SuccessCt>cacheMap[msg.Key].Success{
				s := cacheMap[msg.Key]
				s.Success = msg.SuccessCt
				cacheMap[msg.Key] = s
			}
			if len(Infos) >= batchNum {
				tmpInfos := make([]NeedSyncPrgInfo,len(Infos))
				copy(tmpInfos,Infos)
				Infos = Infos[:0]
				sync(tmpInfos)
			}
		}

		log.Printf("last sync infos %+v",Infos)
		sync(Infos)
		for i:=0;i<3; i++{
			err := g.cache.SyncGroupState(g.groupId,GroupInfo{
				State:  syncInfo.State,
				Reason: syncInfo.FailReason,
				Data:   cacheMap,
			})
			if err == nil {
				break
			}
			time.Sleep(time.Millisecond*200)
		}
		close(syncFinish)
	}()
	return syncFinish
}

func (g *Group) do(group GroupReq) (){
	if group.LoadUserFun == nil {
		g.failGroup("loadUserfun is nil")
		log.Printf("groupId(%d),loadUserfun is nil",group.Id)
		return
	}
	if group.SyncPrgFun == nil {
		g.failGroup("syncPrgFun is nil")
		log.Printf("groupId(%d),syncPrgFun is nil",group.Id)
		return
	}
	ctx,cancel := context.WithCancel(context.Background())
	g.progress = NewProgress(ctx,100)

	//尝试从获取历史记录，如果有发送历史，需要同步状态。
	ginfo, err := g.cache.GetGroupState(group.Id)
	if err != nil {
		log.Printf("groupId(%d),err(%v) groupState error",g.groupId,err)
		return
	}
	if ginfo.State == STATE_FINISH {
		log.Printf("load group(%d),info(%+v) has finish",group.Id,ginfo)
		return
	}
	log.Printf("load group(%d),info(%+v)",group.Id,ginfo)
	if ginfo.State != STATE_INIT {
		g.state = STATE_PROCESS
		g.stopReason = ginfo.Reason
		g.restart = true
	}
	if len(ginfo.Data)>0 {
		for key,val := range ginfo.Data {
			g.progress.AddProgressKey(key)
			g.progress.SetProgressState(key,NeedSyncPrgInfo{
				Key:         key,
				SuccessCt:   val.Success,
				FailCt:      val.Fail,
			})
			fmt.Printf("***********progress %s %+v \n",key,g.progress.info[key])

		}
	}

	//触达场景内容
	eve := event.NewEvent()
	data,err := eve.GetContentBySceneId(group.SceneId)
	if err != nil {
		g.failGroup(err.Error())
		log.Printf("group(%v),get sceneId err(%s)",group,err.Error())
		fmt.Println(err)
		return
	}
	eventMapString,_ := eve.TransferMapStringString(data)
	var smsChan chan SMSInfo
	var smsInfo SMSInfo
	var wxChan chan WxInfo
	var wxInfo WxInfo
	var appChan chan AppInfo
	var appInfo AppInfo
	/*************初始化需要的消息通过，启动对应的goroutine*******************/
	for _,vals := range eventMapString{
		switch vals["msgType"] {
		case MSGTYPE_WX:
			log.Printf("start groupId(%d),ScenId(%d) WxJob",group.Id,group.SceneId)
			wxChan = make(chan WxInfo,g.config.Group.WxMsgWorker+1)
			templateId,_:= eve.GetWxTemplateId(vals["signatureId"])
			keyVals := eve.GetWxTemplateKeyVal(vals)

			wxKV := make(map[string]pkgNotify.Keyworkd,0)
			var color string
			var ok bool
			for k,v := range keyVals{
				if strings.Contains(k, "_color") {
					continue
				}
				ckey := k+"_color"
				if color,ok = keyVals[ckey]; !ok {
					color = ""
				}
				wxKV[k] = pkgNotify.NewKeyword(v,color)
			}
			wxInfo = WxInfo{
				ToUsers:    nil,
				TemplateId: templateId,
				Url:        vals["url"],
				WxKV:       wxKV,
			}
			g.progress.AddProgressKey(WX_KEY)
			g.wxJob(g.config.Group.WxMsgWorker,wxChan)
		case MSGTYPE_APP:
			log.Printf("start groupId(%d),ScenId(%d) AppJob",group.Id,group.SceneId)
			appChan = make(chan AppInfo,g.config.Group.AppMsgQueue)
			appInfo = AppInfo{
				Alias: nil,
				Title: vals["title"],
				Msg:   vals["msg"],
				Url:   vals["url"],
			}
			g.progress.AddProgressKey(APP_KEY)
			g.appJob(g.config.Group.AppMsgWorker,appChan)
		case MSGTYPE_SMS:
			log.Printf("start groupId(%d),ScenId(%d) smsJob",group.Id,group.SceneId)
			smsChan = make(chan SMSInfo,g.config.Group.SmsMsgQueue)
			smsInfo = SMSInfo{
				Mobiles: nil,
				Msg:     vals["msg"],
				Sign:    vals["signatureId"],
			}
			g.progress.AddProgressKey(SMS_KEY)
			g.smsJob(g.config.Group.SmsMsgWorker,smsChan)
		}
	}
	/***************加载待发送的用户********************/
	clients,err := group.LoadUserFun()
	if err != nil {
		g.failGroup("load userInfos fail")
		log.Printf("group(%v),load userInfos fail(%s)",group,err.Error())
		cancel()
		return
	}


	g.startGroup() //更新Group状态

	/**************开启信息同步，主要目的是同步给业务方进度******************/
	syncFinish := g.syncPrg(group.SyncPrgFun)

	/*******读取用户信息，进行用户分组，推送到clientQueue。*****************/
	maxClientNum := g.config.Group.BatchNum
	g.wg.Add(1)
	clientQueue := make(chan []UserInfo,g.config.Group.WxUserInfoQueue)
	go func() {
		defer func() {
			log.Printf("end groupId(%d),ScenId(%d)  push user to client queue ",group.Id,group.SceneId)
			close(clientQueue)
			defer g.wg.Done()
		}()
		log.Printf("start groupId(%d),ScenId(%d) push user to client queue ",group.Id,group.SceneId)
		waitSendClient := make([]UserInfo,0,maxClientNum)
		count := 0
		for _,client := range clients{
			select {
			case <-g.stop:
				log.Printf("end groupId(%d),ScenId(%d) push user to client queue  ",group.Id,group.SceneId)
				waitSendClient = waitSendClient[:0]
				return
			default:
			}
			count++
			waitSendClient = append(waitSendClient,client)
			if count<maxClientNum {
				continue
			}
			dst := make([]UserInfo,len(waitSendClient))
			copy(dst,waitSendClient)
			clientQueue<-dst
			count = 0
			waitSendClient = waitSendClient[:0]
		}
		//将最后一组消息推送
		clientQueue<-waitSendClient
		waitSendClient = nil

	}()



	/************clientQueue 推送到各种消息的消费队列中*************/
	workerWg := sync.WaitGroup{}
	groupWorkerNum := g.config.Group.UserPushWorker
	workerWg.Add(groupWorkerNum)

	for i := 0; i < groupWorkerNum; i++ {
		go func() {
			defer workerWg.Done()
			log.Printf("start groupId(%d),ScenId(%d),worker(%d) push msg ",group.Id,group.SceneId,i)
			for{
				select {
				case wclients,ok :=<-clientQueue:
					if appChan != nil {
						appInfo.Alias = getAppUser(wclients)
						appChan<-appInfo
					}
					if smsChan != nil {
						smsInfo.Mobiles = getMobile(wclients)
						smsChan<-smsInfo
					}
					if wxChan != nil {
						wxInfo.ToUsers = getOpenId(wclients)
						wxChan<-wxInfo
					}
					if len(clientQueue)==0 && !ok {
						log.Printf("end groupId(%d),ScenId(%d),worker(%d)  push msg ",group.Id,group.SceneId,i)
						return
					}
				case <-g.stop:
					log.Printf("stop groupId(%d),ScenId(%d),worker(%d)  push msg ",group.Id,group.SceneId,i)
					return
				}
			}
		}()
	}
	/***************等待推送完成，然后关闭通道******************/
	workerWg.Wait()
	if appChan != nil {
		close(appChan)
	}
	if smsChan != nil {
		close(smsChan)
	}
	if wxChan != nil {
		close(wxChan)
	}
	/*******************等待所有goroutine************************/
	g.wg.Wait()
	/*************************关闭同步任务**********************************/
	g.progress.Close()
	//完成
	g.finishGroup()
	/*************************等待同步完成**********************************/
	<-syncFinish
	log.Printf("finish groupId(%d),ScenId(%d)",group.Id,group.SceneId)
}
type AppInfo struct {
	Alias map[int]string
	Title string
	Msg string
	Url string
}
func (g *Group) getNoSendUser(key string,data map[int]string) []string{
	users := make([]string,0,len(data))
	gkey := getSendedKey(g.groupId,key)
	for pos,user := range data {
		if g.restart == false {
			ok,err := g.cache.SetSended(gkey,int32(pos))
			if ok{
				users = append(users,user)
			}
			if err != nil {
				log.Printf("setSend err(%v)",err)
			}
			continue
		}
		ok,err := g.cache.HasSended(gkey,int32(pos))
		if !ok {
			ok,err = g.cache.SetSended(gkey,int32(pos))
			if ok{
				users = append(users,user)
			}
			if err != nil {
				log.Printf("setSend err(%v)",err)
			}
			users = append(users,user)
		}
	}
	return users
}
func (g *Group) appJob(num int ,ch chan AppInfo){
	g.wg.Add(num)
	for i := 0; i < num; i++ {
		go func() {
			defer g.wg.Done()
			for{
				select {
				case <-g.stop:
					return
				case sendInfo,ok := <-ch :
					if len(sendInfo.Alias)>0 {
						alias := g.getNoSendUser(APP_KEY,sendInfo.Alias)
						if len(alias)<1{
							continue
						}
						msg := pkgNotify.NewAppMsg(alias,sendInfo.Title,sendInfo.Msg,sendInfo.Url)
						if g.config.Debug{
							log.Printf("debug send app msg (%v) \n",msg)
							time.Sleep(time.Microsecond*200)
						} else {
							g.sender.SendApp("xysjiguang",msg)
						}
						g.incrSuccessCounter(APP_KEY,alias)
					}
					if len(ch)==0 && !ok {
						log.Printf("end  app msg send job \n")
						return
					}
				}
			}

		}()
	}
}
type WxInfo struct {
	ToUsers map[int]string
	TemplateId string
	Url string
	WxKV map[string]pkgNotify.Keyworkd
}

func (g *Group) wxPushMsgJob(msgChan chan <-pkgNotify.TemplateMsg,ch <-chan WxInfo)  {
	g.wg.Add(1)
	go func() {
		defer func() {
			close(msgChan)
			defer g.wg.Done()
		}()
		for{
			select {
			case <-g.stop:
				return
			case sendInfo,ok := <-ch :
				tousers := g.getNoSendUser(WX_KEY,sendInfo.ToUsers)
				for _,touser := range tousers {
					if touser == "" {
						continue
					}
					msg := pkgNotify.NewTemplateMsg(touser,sendInfo.TemplateId,sendInfo.Url,sendInfo.WxKV)
					select {
					case msgChan<-msg:
					case <-g.stop:
						return
					}
				}
				if len(ch)==0 && !ok {
					return
				}
			}
		}
	}()
}
func (g *Group)wxJob(num int ,ch chan WxInfo){
	msgChan := make(chan pkgNotify.TemplateMsg,g.config.Group.WxMsgQueue)
	g.wxPushMsgJob(msgChan,ch)
	/**********************调用微信接口，发送微信消息*************************/
	g.wg.Add(num)
	for i := 0; i < num; i++ {
		go func(workerNum int) {
			defer func() {
				log.Printf("end  wx msg send job \n")
				g.wg.Done()
			}()
			for  {
				select {
				case msg,ok := <-msgChan:
					if msg.Touser != "" {
						if g.config.Debug{
							log.Printf("debug send wx msg (%+v),ok(%v), i(%d) \n",msg,ok,workerNum)
							time.Sleep(time.Microsecond*200)
						} else {
							g.sender.SendWxTemplate("xiaoyusan",msg)
						}
						g.incrSuccessCounter(WX_KEY,[]string{msg.Touser})
					}

					if !ok && len(msgChan)<1{
						return
					}
				}
			}

		}(i)
	}
}
type SMSInfo struct {
	Mobiles map[int]string
	Msg string
	Sign string
}
func (g *Group)smsJob(num int ,ch chan SMSInfo){
	g.wg.Add(num)
	for i := 0; i < num; i++ {
		go func() {
			defer func() {
				g.wg.Done()
				log.Printf("end  sms msg send job")
			}()

			for{
				select {
				case <-g.stop:
					return
				case sendInfo,ok := <-ch:
					if len(sendInfo.Mobiles)>0 {
						mobilesArr := g.getNoSendUser(SMS_KEY,sendInfo.Mobiles)
						if len(mobilesArr)<1 {
							continue
						}
						mobiles := strings.Join(mobilesArr,",")
						msg := pkgNotify.NewSmsMsg(mobiles,sendInfo.Msg,sendInfo.Sign)
						if g.config.Debug {
							log.Printf("debug send sms msg (%v) \n",msg)
							time.Sleep(time.Microsecond*200)
						} else {
							g.sender.SendSms("chuanglan",msg)
						}
						g.incrSuccessCounter(SMS_KEY,mobilesArr)
					}
					if len(ch)==0 && !ok {
						return
					}

				}
			}
		}()
	}
}

func getSendedKey(groupId int64,key string) string {
	return fmt.Sprintf("group:%d:%s",groupId,key)
}
